package com.lagou.rpc.client;

import com.lagou.rpc.entity.RpcRequest;
import com.lagou.rpc.entity.RpcResponse;
import io.netty.channel.*;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * 客户端处理类，异步处理netty返回结果。
 */
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {

    private volatile Channel channel;

    private Map<String, RpcFuture> futureMap = new ConcurrentHashMap<>();

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {
        RpcFuture rpcFuture = futureMap.get(msg.getRequestId());
        if (rpcFuture != null) {
            futureMap.remove(msg.getRequestId());
            rpcFuture.doReceived(msg);
        }
    }

    public RpcFuture sendRequest(RpcRequest request) {
        final CountDownLatch latch = new CountDownLatch(1);
        RpcFuture rpcFuture = new RpcFuture();
        futureMap.put(request.getRequestId(), rpcFuture);
        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                latch.countDown();
            }
        });
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return rpcFuture;
    }

}
